package com.tanjor.kafka.listener;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.collections.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.listener.MessageListener;

import com.tanjor.kafka.KafkaContainer;
import com.tanjor.kafka.SpringUtil;
import com.tanjor.kafka.annotation.KafkaMsgHander;
import com.tanjor.kafka.consumer.KafkaConsumerService;


/**
 * kafka监听器启动
 * 自动监听是否有消息需要消费
 * @author wangb
 *
 */
public class KafkaConsumerListener implements MessageListener<String, String>{
	 protected final Logger LOG = LoggerFactory.getLogger("kafkaConsumer");

	 
	 private static Map<String,List<KafkaContainer>> msgContainer = new HashMap<String, List<KafkaContainer>>();
	
	 static{
		 Map<String, Object> serviceBeanMap = SpringUtil.getApplicationContext().getBeansWithAnnotation(KafkaMsgHander.class);
         for (Object serviceBean : serviceBeanMap.values()) {
        	 String name = serviceBean.getClass().getAnnotation(KafkaMsgHander.class).name();
        	 String topic = serviceBean.getClass().getAnnotation(KafkaMsgHander.class).topic();
        	 KafkaConsumerService kafkaConsumerService = (KafkaConsumerService) serviceBean;
        	 List<KafkaContainer> list = msgContainer.get(topic);
        	 if(CollectionUtils.isEmpty(list)){
        		 list = new ArrayList<KafkaContainer>();
        	 }
        	 KafkaContainer kafkaContainer = new KafkaContainer();
        	 kafkaContainer.setKafkaConsumerService(kafkaConsumerService);
        	 kafkaContainer.setName(name);
        	 kafkaContainer.setTopic(topic);
        	 list.add(kafkaContainer);
        	 msgContainer.put(topic, list);
         }
	 }
	 
	 /**
	 * 监听器自动执行该方法
	 * 	消费消息
	 * 	自动提交offset
	 * 	执行业务代码
	 * 	（high level api 不提供offset管理，不能指定offset进行消费）
	 */
	@Override
	public void onMessage(ConsumerRecord<String, String> record) {
		LOG.info("=============kafkaConsumer开始消费=============");
		String topic = record.topic();
		String key = record.key();
		String value = record.value();
		long offset = record.offset();
		int partition = record.partition();
		
		List<KafkaContainer> kafkaMsgs = msgContainer.get(topic);
		if(CollectionUtils.isNotEmpty(kafkaMsgs)){			
			for(KafkaContainer KafkaContainer:kafkaMsgs){
				new KafkaThread(KafkaContainer,value).start();
			}
		}else{
			LOG.info("-------------该服务中没有topic为"+topic+"的消费者。");
		}
		LOG.info("-------------topic:"+topic);
		LOG.info("-------------value:"+value);
		LOG.info("-------------key:"+key);
		LOG.info("-------------offset:"+offset);
		LOG.info("-------------partition:"+partition);
		LOG.info("~~~~~~~~~~~~~kafkaConsumer消费结束~~~~~~~~~~~~~");
	}
	
	 
	class KafkaThread extends Thread {
		
		private KafkaContainer kafkaContainer;
		private String value;
		
	    public KafkaThread(KafkaContainer kafkaContainer,String value) {
	       this.kafkaContainer = kafkaContainer;
	       this.value = value;
	    }
    
	    @Override
	    public void run() {
	    	try{
	    		kafkaContainer.getKafkaConsumerService().prosessMsg(this.value);
	    	}catch(Exception e){
	    		LOG.info("============="+KafkaConsumerService.class+":"+kafkaContainer.getName()+"执行失败============="+e.getMessage());
	    	}
	    }
	}

}
